package com.sun.clients.kafka;

import com.sun.clients.properties.AppProperty;
import com.sun.clients.service.MessageService;
import com.sun.clients.task.MessageTask;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public class ConsumerExecutor {

    private KafkaConsumer<String, String> kafkaConsumer;
    private ExecutorService executorService;
    private MessageService messageService;
    private AppProperty appProperty;

    public ConsumerExecutor(KafkaConsumer<String, String> kafkaConsumer,
                            ExecutorService executorService,
                            MessageService messageService,
                            AppProperty appProperty) {
        this.kafkaConsumer = kafkaConsumer;
        this.executorService = executorService;
        this.messageService = messageService;
        this.appProperty = appProperty;

        this.kafkaConsumer.subscribe(Arrays.asList(appProperty.getTopics().split(",")));
    }

    /**
     * 消费消息
     */
    public void execute() {
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(2000L));
            if(!records.isEmpty()) {
                try {
                    executorService.execute(new MessageTask(messageService, records));
                } catch (Exception e) {
                    log.error("consumer execute error", e);
                }
            }
        }
    }

    /**
     * 关闭
     */
    public void shutdown() {
        if(kafkaConsumer != null) {
            kafkaConsumer.close();
        }
        if(executorService != null) {
            executorService.shutdown();
        }
        try {
            if(executorService != null && executorService.awaitTermination(appProperty.getAwaitTerminationTime(), TimeUnit.SECONDS)) {
                log.error("execute timeout");
            }
        } catch (InterruptedException e) {
            log.error("InterruptedException", e);
            Thread.currentThread().interrupt();
        }
    }

}
